package spring.cloud.service.task.service.impl;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.StopWatch;
import spring.cloud.service.task.domain.different_info.DifferentType;
import spring.cloud.service.task.domain.different_info.TaskConfigurationDifferentInfo;
import spring.cloud.service.task.domain.different_info.TaskDifferentInfo;
import spring.cloud.service.task.domain.task.ZooKeeperTaskConfigurationTransferUtil;
import spring.cloud.service.task.dto.TaskDto;
import spring.cloud.service.task.mapper.NameSpaceDtoMapper;
import spring.cloud.service.task.mapper.TaskDtoMapper;
import spring.cloud.service.task.service.TaskService;
import spring.cloud.service.task.service.ZooKeeperDatabaseTaskDifferentInfoService;
import spring.cloud.service.task.service.ZooKeeperService;
import spring.cloud.tasks.common.ThreadFactory;
import spring.cloud.tasks.common.curator.Curator;
import spring.cloud.tasks.common.nodepaths.TaskBaseInfoNode;
import spring.cloud.tasks.common.nodepaths.TaskNodePaths;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

@Slf4j
public class ZooKeeperDatabaseTaskDifferentInfoServiceImpl implements ZooKeeperDatabaseTaskDifferentInfoService {


    private static final int DIFF_THREAD_NUM = 10;

    @Autowired
    private NameSpaceDtoMapper nameSpaceDtoMapper;
    @Resource
    private TaskDtoMapper taskConfigurationDtoMapper;
    @Resource
    private ZooKeeperService zooKeeperService;
    @Resource
    private TaskService taskService;

    private ExecutorService executorService;

    @PostConstruct
    public void init() {
        if (executorService != null) {
            executorService.shutdownNow();
        }
        executorService = Executors.newFixedThreadPool(DIFF_THREAD_NUM, new ThreadFactory("diff-zk-db-thread", false));
    }

    @PreDestroy
    public void destroy() {
        if (executorService != null) {
            executorService.shutdownNow();
        }
    }

    @Override
    public List<TaskDifferentInfo> differentByZooKeeperClusterKey(String zooKeeperKey) throws RuntimeException {

        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        List<String> nameSpaceList = nameSpaceDtoMapper.getNamespaceKeyListByZookeeperKey(zooKeeperKey);
        List<Callable<List<TaskDifferentInfo>>> callableList = new ArrayList<>();
        for (final String nameSpace : nameSpaceList) {
            Callable<List<TaskDifferentInfo>> callable = new Callable<List<TaskDifferentInfo>>() {
                @Override
                public List<TaskDifferentInfo> call() throws Exception {
                    return differentByNameSpace(nameSpace);
                }
            };
            callableList.add(callable);
        }

        List<TaskDifferentInfo> taskDifferentInfoList = new ArrayList<>();
        try {
            List<Future<List<TaskDifferentInfo>>> futureList = executorService.invokeAll(callableList);
            for (Future<List<TaskDifferentInfo>> future : futureList) {
                List<TaskDifferentInfo> list = future.get();
                if (list != null && !list.isEmpty()) {
                    taskDifferentInfoList.addAll(list);
                }
            }
        } catch (InterruptedException e) {// NOSONAR
            log.warn("the thread is interrupted", e);
            throw new RuntimeException("the diff thread is interrupted", e);
        } catch (Exception e) {
            log.error("exception happens during execute diff operation", e);
            throw new RuntimeException(e);
        }
        stopWatch.stop();
        log.info("Finish diff zkcluster:{}, which cost {}ms", zooKeeperKey, stopWatch.getTotalTimeSeconds());
        return taskDifferentInfoList;
    }


    @Override
    public List<TaskDifferentInfo> differentByNameSpace(String nameSpace) throws RuntimeException {
        StopWatch stopWatch = new StopWatch();
        List<TaskDifferentInfo> taskDifferentInfoList = new ArrayList<>();
        try {
            stopWatch.start();
            List<TaskDto> taskDtoList = taskConfigurationDtoMapper.getTaskDtoListByNamespaceKey(nameSpace);
            if (taskDtoList == null || taskDtoList.isEmpty()) {
                return taskDifferentInfoList;
            }
            Curator curator = ZooKeeperServiceImpl.getCurator(zooKeeperService, nameSpace);
            if (curator == null) {
                return taskDifferentInfoList;
            }
            for (TaskDto taskDto : taskDtoList) {
                String taskId = taskDto.getTaskId();
                if (!checkExists(curator, taskId)) {
                    taskDifferentInfoList.add(new TaskDifferentInfo(nameSpace, taskId, DifferentType.DATABASE_ONLY, new ArrayList<>()));
                    continue;
                }
                TaskDto taskConfigurationFromZooKeeper = ZooKeeperTaskConfigurationTransferUtil.getTaskConfigurationFromZooKeeper(curator, taskId);
                TaskDto taskConfiguration = new TaskDto();
                TaskDifferentInfo taskDifferentInfo = different(nameSpace, taskConfiguration, taskConfigurationFromZooKeeper);
                if (taskDifferentInfo != null) {
                    taskDifferentInfoList.add(taskDifferentInfo);
                }
            }
            Set<String> taskIdSetInDb = taskDtoList.stream().map(taskConfigurationDto -> {
                return taskConfigurationDto.getTaskId();
            }).collect(Collectors.toSet());

            List<String> taskIdList = taskService.getTaskIdListFromZooKeeper(nameSpace);
            for (String taskId : taskIdList) {
                if (!taskIdSetInDb.contains(taskId)) {
                    taskDifferentInfoList.add(new TaskDifferentInfo(nameSpace, taskId, DifferentType.ZOOKEEPER_ONLY, new ArrayList<>()));
                }
            }
        } catch (RuntimeException e) {
            log.error(e.getMessage(), e);
            throw e;
        } catch (Exception e) {
            log.error("exception throws during diff by nameSpace [{}]", nameSpace, e);
            throw new RuntimeException(e);
        } finally {
            stopWatch.stop();
            log.info("Finish diff nameSpace:{} which cost {}s", nameSpace, stopWatch.getTotalTimeSeconds());
        }

        return taskDifferentInfoList;
    }


    @Override
    public TaskDifferentInfo differentByTaskId(String nameSpace, String taskId) throws RuntimeException {
        Curator curator;
        try {
            curator = ZooKeeperServiceImpl.getCurator(zooKeeperService, nameSpace);
            if (curator == null) {
                return null;
            }
            log.info("start to diff job:{}", taskId);

            TaskDto taskDto = taskConfigurationDtoMapper.getTaskDtoByNamespaceKeyAndTaskId(nameSpace, taskId);
            TaskDto taskConfigurationDto = new TaskDto();
            TaskDto taskConfigurationFromZooKeeper = ZooKeeperTaskConfigurationTransferUtil.getTaskConfigurationFromZooKeeper(curator, taskId);

            if (taskConfigurationDto == null) {
                if (taskConfigurationFromZooKeeper != null) {
                    return new TaskDifferentInfo(nameSpace, taskId, DifferentType.ZOOKEEPER_ONLY, new ArrayList<>());
                } else {
                    return null;
                }
            } else {
                if (taskConfigurationFromZooKeeper == null) {
                    return new TaskDifferentInfo(nameSpace, taskId, DifferentType.DATABASE_ONLY, new ArrayList<>());
                } else {
                    // diff only when dbJobConfig and zkJobConfig both not null
                    return different(nameSpace, taskConfigurationDto, taskConfigurationFromZooKeeper);
                }
            }
        } catch (Exception e) {
            log.error("exception throws during diff by nameSpace [{}] and job [{}]", nameSpace, taskId, e);
            throw new RuntimeException(e);
        }
    }

    private boolean checkExists(Curator curator, String taskId) {
        return curator.checkExists(TaskNodePaths.getTaskNodePath(taskId));
    }

    /**
     * zk中的任务配置和db中的任务配置的对比。
     *
     * @param dbTaskConfiguration db里面的配置。
     * @param zkTaskConfiguration zk里面的配置
     */
    protected TaskDifferentInfo different(String nameSpace, TaskDto dbTaskConfiguration, TaskDto zkTaskConfiguration) {
        String taskId = dbTaskConfiguration.getTaskId();

        List<TaskConfigurationDifferentInfo> list = new ArrayList<>();

        different(TaskBaseInfoNode.task_type, dbTaskConfiguration.getTaskType(), zkTaskConfiguration.getTaskType(), list);
        //task_id
        different(TaskBaseInfoNode.xid, dbTaskConfiguration.getXid(), zkTaskConfiguration.getXid(), list);
        different(TaskBaseInfoNode.description, dbTaskConfiguration.getDescription(), zkTaskConfiguration.getDescription(), list);
        //
        different(TaskBaseInfoNode.cron_expression, dbTaskConfiguration.getCronExpression(), zkTaskConfiguration.getCronExpression(), list);
        different(TaskBaseInfoNode.pause_period_date, dbTaskConfiguration.getPausePeriodDate(), zkTaskConfiguration.getPausePeriodDate(), list);
        different(TaskBaseInfoNode.pause_period_time, dbTaskConfiguration.getPausePeriodTime(), zkTaskConfiguration.getPausePeriodTime(), list);
        //
        different(TaskBaseInfoNode.timeout_minutes, String.valueOf(dbTaskConfiguration.getTimeoutMinutes()), String.valueOf(zkTaskConfiguration.getTimeoutMinutes()), list);
        different(TaskBaseInfoNode.timeout_alarm_minutes, String.valueOf(dbTaskConfiguration.getTimeoutAlarmMinutes()), String.valueOf(zkTaskConfiguration.getTimeoutAlarmMinutes()), list);
        //
        //
        //
        different(TaskBaseInfoNode.task_properties, dbTaskConfiguration.getTaskProperties(), zkTaskConfiguration.getTaskProperties(), list);
        different(TaskBaseInfoNode.task_data_rang_rule, dbTaskConfiguration.getTaskDataRangeRule(), zkTaskConfiguration.getTaskDataRangeRule(), list);
        different(TaskBaseInfoNode.task_item_info, dbTaskConfiguration.getTaskItemInfo(), zkTaskConfiguration.getTaskItemInfo(), list);
        //
        different(TaskBaseInfoNode.status, String.valueOf(dbTaskConfiguration.getStatus()), String.valueOf(zkTaskConfiguration.getStatus()), list);
        different(TaskNodePaths.prefer_executor_id_list, dbTaskConfiguration.getPreferExecutorIdList(), zkTaskConfiguration.getPreferExecutorIdList(), list);
        if (!list.isEmpty()) {
            return new TaskDifferentInfo(nameSpace, taskId, DifferentType.DATABASE_ZOOKEEPER_HAS_DIFFERENCE, list);
        } else {
            return null;
        }
    }

    public void different(String key, String valueInDb, String valueInZk, List<TaskConfigurationDifferentInfo> taskConfigurationDifferentInfoList) {
        // 这里处理所有valueInDB 为空的情况
        if (StringUtils.isEmpty(valueInDb)) {
            if (StringUtils.isEmpty(valueInZk)) {
                return;
            } else {
                log.debug("key:{} has difference between zk and db", key);
                taskConfigurationDifferentInfoList.add(new TaskConfigurationDifferentInfo(key, valueInDb, valueInZk));
                return;
            }

        } else {
            // valueInDB != null && valueInZk == null
            if (StringUtils.isEmpty(valueInZk)) {
                log.debug("key:{} has difference between zk and db", key);
                taskConfigurationDifferentInfoList.add(new TaskConfigurationDifferentInfo(key, valueInDb, valueInZk));
                return;
            }
            // 处理非String类型
            if (!valueInDb.trim().equals(valueInZk.trim())) {
                log.debug("key:{} has difference between zk and db", key);
                taskConfigurationDifferentInfoList.add(new TaskConfigurationDifferentInfo(key, valueInDb, valueInZk));
                return;
            } else {
                return;
            }
        }


    }


}
